﻿using System;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Linq.Expressions;
using System.Text;
using System.Threading;
using NUnit.Framework;

namespace Universe.SimpleQueue.Tests
{
    [Explicit]
    public class StressDefinitions
    {
        private const string QueueName = "Stress Queue";

        protected virtual Func<SimpleQueue.IConnectionInfo> ConnectionFactory
        {
            get
            {
                return Env.SqlServer_With_ReadCommitted_Isolated_Transaction;
            }
        } 

        private Action<SimpleQueue>[] SmallPublishers = new[]
        {
            TypicalWorkers.PublishMessage(QueueName, 1, 11),
            TypicalWorkers.PublishNewMessage(QueueName, 1, 11),
            TypicalWorkers.PublishNewMessage_WithCheck(QueueName, 1, 11),
            TypicalWorkers.PublishSameMessage(QueueName, 1),
            TypicalWorkers.PublishSameMessage_WithCheck(QueueName, 1),
        };

        private Action<SimpleQueue>[] LargePublishers = new[]
        {
            TypicalWorkers.PublishMessage(QueueName, 10000000, 10000000 + 11111),
            TypicalWorkers.PublishNewMessage(QueueName, 10000000, 10000000 + 11111),
            TypicalWorkers.PublishNewMessage_WithCheck(QueueName, 10000000, 10000000 + 11111),
            TypicalWorkers.PublishSameMessage(QueueName, 10000000),
            TypicalWorkers.PublishSameMessage_WithCheck(QueueName, 10000000),
        };


        private Action<SimpleQueue>[] ConsumersAndProducers = new[]
        {
            TypicalWorkers.Consume_And_Forward_But_Sometimes_Fail(QueueName, "No Way", 1m), 
            TypicalWorkers.Consume_But_Sometimes_Fail(QueueName, 0.1m), 
            TypicalWorkers.Consume_But_Sometimes_Fail(QueueName, 0.9m), 
            TypicalWorkers.Consume_And_Always_Ack(QueueName), 
            TypicalWorkers.ReadFromEmptyQueue(),
            
            TypicalWorkers.PublishMessage(QueueName, 1, 11),
            TypicalWorkers.PublishNewMessage(QueueName, 1, 11),
            TypicalWorkers.PublishNewMessage_WithCheck(QueueName, 1, 11),

            TypicalWorkers.Consume_And_Forward_But_Sometimes_Fail(QueueName, "No Way", 1m), 
            TypicalWorkers.Consume_But_Sometimes_Fail(QueueName, 0.1m), 
            TypicalWorkers.Consume_And_Always_Ack(QueueName), 
            TypicalWorkers.ReadFromEmptyQueue(),

        };

        [Test]
        public void ManyWorkers([Values(2)] int duration)
        {
            SimpleQueue queue = new SimpleQueue(ConnectionFactory);

            using(var con = queue.ConnectionFactory().Connection)
            Trace.WriteLine("DB is " + con.GetType().Name);

            Trace.WriteLine("PRODUCING SHORT MESSAGES DURING " + duration + " SECS");
            var all = SmallPublishers;
            for (int n = 0; n < all.Count(); n++)
            {
                var arg = all.Take(n+1).ToArray();
                Run(duration*1000L, queue, arg);
            }

            Trace.WriteLine("CONSUMERS & PRODUCERS DURING " + duration + " SECS");
            all = ConsumersAndProducers;
            for (int n = 0; n < all.Count(); n++)
            {
                var arg = all.Take(n + 1).ToArray();
                Run(duration * 1000L, queue, arg);
            }

            Trace.WriteLine("PRODUCING LONG MESSAGES  DURING " + duration + " SECS");
            all = LargePublishers;
            for (int n = 0; n < all.Count(); n++)
            {
                var arg = all.Take(n + 1).ToArray();
                Run(duration * 1000L, queue, arg);
            }

            Trace.WriteLine(SimpleQueue.LocalStat.AsConsoleTable());

        }

        void Run(long duration, SimpleQueue queue, Action<SimpleQueue>[] actions)
        {
            // Heat JIT & SQL Server, ignore exception
            foreach (var action in actions)
            {
                try
                {
                    action(queue);
                }
                catch (Exception ex)
                {
                }
            }


            Dictionary<int,object> totalThreads = new Dictionary<int, object>();
            long totalActions = 0;
            Barrier b = new Barrier(actions.Count());
            List<Thread> threads = new List<Thread>();
            FailCollection fails = new FailCollection();
            int[] numbers = new int[actions.Count()];
            int index = 0;
            foreach (var action in actions)
            {
                var a = action;
                int current = index;
                Thread t = new Thread(() =>
                {
                    lock (totalThreads) totalThreads[Thread.CurrentThread.ManagedThreadId] = null;

                    int n1 = 0;
                    b.SignalAndWait();
                    Stopwatch sw = Stopwatch.StartNew();
                    do
                    {
                        try
                        {
                            a(queue);
                            Interlocked.Increment(ref totalActions);
                            n1++;
                        }
                        catch (Exception ex)
                        {
                            FailInfo fi = new FailInfo(ex, current);
                            fails.SmartAdd(fi);
                            DumpException("Stress Action failed: " + ex);
                        }

                    } 
                    while (sw.ElapsedMilliseconds < duration);
                    
                    lock (numbers) numbers[current] = n1;
                    b.SignalAndWait();

                }) { IsBackground = true};

                t.Start();
                threads.Add(t);
                index++;
            }

            foreach (var thread in threads)
                thread.Join();

            string totalActionsAs =
                string.Join("+", numbers.Select(x => x.ToString()))
                                 + " = " + totalActions;

            if (actions.Count() == 1)
                totalActionsAs = totalActions.ToString();

            
            Trace.WriteLine(string.Format(
                "Workers: {0}. Total actions: {1}, threads: {2}", 
                actions.Count(),
                totalActionsAs, 
                totalThreads.Count));

            if (fails.Count > 0)
            {
                var asStr = fails
                    .OrderByDescending(x => x.Count)
                    .Select(x => string.Format("   :) #{0}, {1} times {2}", x.Worker, x.Count, x.Reason));

                Trace.WriteLine(string.Join(Environment.NewLine, asStr));
            }

        }

        static readonly StringBuilder Exceptions = new StringBuilder();
        private void DumpException(string s)
        {

            lock (Exceptions)
            {
                if (Exceptions.Length == 0)
                    AppDomain.CurrentDomain.DomainUnload += (sender, args) =>
                    {
                        File.WriteAllText(
                            string.Format("Exceptions-{0}.txt", DateTime.Now.ToString("yyyy-MM-dd HH-mm-ss")),
                            Exceptions.ToString());
                    };

                Exceptions.AppendLine().AppendLine(s);
            }


        }

        class FailInfo
        {
            public string AsString;
            public int Worker;
            public string Reason;
            public int Count;

            public FailInfo(Exception ex, int worker)
            {
                Worker = worker;
                Count = 1;
                AsString = ex.ToString();

                List<string> msgs = new List<string>();
                while (ex != null)
                {
                    msgs.Add(ex.Message);
                    ex = ex.InnerException;
                }

                Reason = String.Join(" -->-- ", msgs);
            }
        }

        class FailCollection : Collection<FailInfo>
        {
            public void SmartAdd(FailInfo fi)
            {
                lock (base.Items)
                {
                    var found = this.FirstOrDefault(x => x.Worker == fi.Worker && x.AsString == fi.AsString);
                    if (found == null)
                        this.Add(fi);

                    else
                        found.Count++;
                }
            }
        }
    }
}
